package com.sinozo.data.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.sinozo.data.app.function.UserActionTimestampsAndWatermarks;
import com.sinozo.data.bean.UserActionBean;
import com.sinozo.data.bean.key.UserActionKeyByBean;
import com.sinozo.data.common.ConfigConstant;
import com.sinozo.data.common.DataConfig;
import com.sinozo.data.utils.*;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @author zhaoyb
 * @version 1.0
 * @description 用户动作
 * @date 2023/5/17 09:38:42
 */
public class DwsTrafficPageActionWindow {


    public static void main(String[] args) {
        Logger logger = LoggerFactory.getLogger(DwsTrafficPageActionWindow.class);
        try {
            //1.获取执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(DataConfig.KAFKA_PARTITION_NUM);

            //1.1 开启CheckPoint
            env.enableCheckpointing(DataConfig.FLINK_CHECKPOINT_INTERVAL, CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(DataConfig.FLINK_CHECKPOINT_TIMEOUT);
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);

            //1.2 设置状态后端
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
            env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
            );
            env.setRestartStrategy(RestartStrategies.failureRateRestart(10, org.apache.flink.api.common.time.Time.of(3L, TimeUnit.DAYS), org.apache.flink.api.common.time.Time.of(1L, TimeUnit.MINUTES)));

            String hdfsPath = DataConfig.HDFS_SERVER;
            env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage(PathUtil.getHdfsPath(hdfsPath,"dwsTrafficPageActionWindow")));
            env.setStateBackend(new EmbeddedRocksDBStateBackend());

            // 2.读取两个主题的数据创建流
            String uvTopic = ConfigConstant.KAFKA_UNIQUE_ACTION_TOPIC;
            String uvEventTopic = ConfigConstant.KAFKA_UNIQUE_EVENT_ACTION_TOPIC;
            String topic = ConfigConstant.KAFKA_ACTION_LOG_TOPIC;
            String groupId = ConfigConstant.KAFKA_USER_ACTION_WINDOW_GROUP_ID;

            DataStreamSource<String> uvDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(uvTopic, groupId));
            DataStreamSource<String> uvEventDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(uvEventTopic, groupId));
            DataStreamSource<String> pageDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

            // 3.统一数据格式
            SingleOutputStreamOperator<UserActionBean> userUniqueActionWithUvDS = uvDS.map(line -> {
                JSONObject jsonObject = JSON.parseObject(line);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject action = jsonObject.getJSONObject("action");

                UserActionBean bean = UserActionBean.builder().appVer(common.getString("app_ver")).os(common.getString("os"))
                        .qid(common.getString("qid")).groupQid(ValueEmptyUtils.emptyString(common.getString("group_qid")))
                        .ascQid(common.getString("asc_qid")).code(common.getString("code"))
                        .isNew(common.getLong("is_new")).event(action.getString("event")).subEvent(action.getString("sub_event"))
                        .labCode(common.getString("lab_code")).labGroupCode(common.getString("lab_group_code"))
                        .action(action.getString("action_type")).actionCnt(0L).deviceCnt(1L).distinctDeviceCnt(0L).ts(action.getLong("ts")).build();
                return bean;
            });

            SingleOutputStreamOperator<UserActionBean> userUniqueEventActionWithUvDS = uvEventDS.map(line -> {
                JSONObject jsonObject = JSON.parseObject(line);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject action = jsonObject.getJSONObject("action");

                UserActionBean bean = UserActionBean.builder().appVer(common.getString("app_ver")).os(common.getString("os"))
                        .qid(common.getString("qid")).groupQid(ValueEmptyUtils.emptyString(common.getString("group_qid"))).ascQid(common.getString("asc_qid")).code(common.getString("code"))
                        .isNew(common.getLong("is_new")).event(action.getString("event")).subEvent(action.getString("sub_event"))
                        .labCode(common.getString("lab_code")).labGroupCode(common.getString("lab_group_code"))
                        .action(action.getString("action_type")).actionCnt(0L).deviceCnt(0L).distinctDeviceCnt(1L)
                        .ts(action.getLong("ts")).build();
                return bean;
            });

            SingleOutputStreamOperator<UserActionBean> userActionWithPageDS = pageDS.map(line -> {
                JSONObject jsonObject = JSON.parseObject(line);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject action = jsonObject.getJSONObject("action");

                UserActionBean bean = UserActionBean.builder().appVer(common.getString("app_ver")).os(common.getString("os"))
                        .qid(common.getString("qid")).groupQid(ValueEmptyUtils.emptyString(common.getString("group_qid")))
                        .ascQid(common.getString("asc_qid")).code(common.getString("code"))
                        .labCode(common.getString("lab_code")).labGroupCode(common.getString("lab_group_code"))
                        .isNew(common.getLong("is_new")).event(action.getString("event")).subEvent(action.getString("sub_event"))
                        .action(action.getString("action_type")).actionCnt(1L).deviceCnt(0L).distinctDeviceCnt(0L).ts(action.getLong("ts")).build();
                return bean;
            });

            // 4.将流进行Union
            DataStream<UserActionBean> unionDS = userActionWithPageDS.union(userUniqueActionWithUvDS, userUniqueEventActionWithUvDS);

            // 5.提取事件时间生成WaterMark
            SingleOutputStreamOperator<UserActionBean> userInstallViewWithWmDS = unionDS.assignTimestampsAndWatermarks(new UserActionTimestampsAndWatermarks());

            // 6.分组开窗聚合
            WindowedStream<UserActionBean, UserActionKeyByBean, TimeWindow> windowedStream = userInstallViewWithWmDS.keyBy(new KeySelector<UserActionBean, UserActionKeyByBean>() {
                @Override
                public UserActionKeyByBean getKey(UserActionBean value) throws Exception {
                    UserActionKeyByBean keyByBean = new UserActionKeyByBean();
                    BeanUtils.copyProperties(keyByBean, value);
                    return keyByBean;
                }
            }).window(TumblingEventTimeWindows.of(DataConfig.WINDOW_TIME));


            SingleOutputStreamOperator<UserActionBean> resultDS = windowedStream.reduce(new ReduceFunction<UserActionBean>() {
                @Override
                public UserActionBean reduce(UserActionBean value1, UserActionBean value2) throws Exception {
                    value1.setActionCnt(value1.getActionCnt() + value2.getActionCnt());
                    value1.setDeviceCnt(value1.getDeviceCnt() + value2.getDeviceCnt());
                    value1.setDistinctDeviceCnt(value1.getDistinctDeviceCnt() + value2.getDistinctDeviceCnt());
                    return value1;
                }
            }, new WindowFunction<UserActionBean, UserActionBean, UserActionKeyByBean, TimeWindow>() {
                @Override
                public void apply(UserActionKeyByBean key, TimeWindow window, Iterable<UserActionBean> input, Collector<UserActionBean> out) throws Exception {
                    //获取数据
                    UserActionBean next = input.iterator().next();
                    //补充信息
                    next.setWinStartTime(DateFormatUtil.toYmdHms(window.getStart()));
                    next.setWinEndTime(DateFormatUtil.toYmdHms(window.getEnd()));
                    //修改TS
                    next.setTs(System.currentTimeMillis());
                    next.setCreateTime(DateFormatUtil.toYmdHms(next.getTs()));
                    //输出数据
                    out.collect(next);
                }
            });

            resultDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_traffic_actions_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

            // 8.启动任务
            env.execute("DwsUserUserActionWindow");
        } catch (Exception e) {
            logger.error("error exec", e);
        }

    }

}
